package com.cskaoyan.order.mq.consumer;
//
//import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.PostConstruct;
//import java.io.UnsupportedEncodingException;
//import java.util.List;
//
///**
// * @Aurhor: 帅的一批
// * @CreateAt: 2022/10/28 21:41
// * @Description:
// * @PACKAGE_NAME： com.cskaoyan.order.mq.consumer
// */
//
//@Component
//public class DelayCancelConsumer {
//
//    private DefaultMQPushConsumer consumer;
//
//    @PostConstruct
//    private void init(){
//        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer");
//
//        consumer.setNamesrvAddr("127.0.0.1:9999");
//
//        consumer.setMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
//                MessageExt messageExt = list.get(0);
//                byte[] body = messageExt.getBody();
//
//                try {
//                    String orderId = new String(body, 0, body.length, "utf-8");
//                    // 订单取消的逻辑
//                    // TODO: 2022/10/28
//                    System.out.println(orderId);
//                    // 1. 判断订单状态，如果是已支付，或者已取消，什么都不做
//                    // 2. 如果订单的状态是，初始化状态，开始执行订单取消:
//                    //    a. 将订单状态修改为已取消
//                    //    b. 根据orderId查询订单中的所有商品条目，根据每一个商品下单的数量 -lock_count  +stock_count
//
//                } catch (UnsupportedEncodingException e) {
//                    e.printStackTrace();
//                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//
//        //订阅主题
//        try {
//            consumer.subscribe("order_delay_cancel","*");
//            consumer.start();
//        } catch (MQClientException e) {
//            e.printStackTrace();
//        }
//    }
//}


import com.cskaoyan.order.dal.entitys.Order;
import com.cskaoyan.order.dal.persistence.OrderMapper;
import com.cskaoyan.order.service.OrderCoreService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

@Slf4j
@Component
public class DelayCancelConsumer {

    @Autowired
    OrderMapper orderMapper;

    @Autowired
    OrderCoreService orderCoreService;

    private DefaultMQPushConsumer consumer;

    @Value("${rocketmq.address}")
    String rocketmqAddress;

    @PostConstruct
    public void init() {
        // 创建consumer对象
        consumer = new DefaultMQPushConsumer("delay_order_cancel_consumer");

        // 设置nameserver地址
        consumer.setNamesrvAddr(rocketmqAddress);

        //设置消息监听器
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                MessageExt message = msgs.get(0);
                byte[] body = message.getBody();
                try {
                    String orderId = new String(body, 0, body.length, "utf-8");
                    log.info(orderId);
                    Order order = orderMapper.selectByPrimaryKey(orderId);
                    //如果订单状态仍然处于初始化状态 就取消订单
                    if (order.getStatus() == 0) {
                        orderCoreService.cancelOrder(orderId);
                    }

                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });


        try {
            // 订阅主题
            consumer.subscribe("order_delay_cancel", "*");

            // 启动consumer
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }

}

